
#include <algorithm>

#include "chordnode.h"
#include "common.h"
#include "simulator.h"
#include "peerloger.h"
#include "application.h"
#include "chordprotocol.h"
#include "chordmsg.h"
#include "eventdefs.h"
#include "chordgod.h"
#include "config.h"

ChordNode::ChordNode( chord_node_id_t c_nid, node_id_t n_id ) 
	: Node(n_id)
	, chord_nid_(c_nid)
	, startup_net_id_(invalid_chord_net_id)
{
	NEW_OBJECT( Application, app_, (this) );
}

ChordNode::~ChordNode()
{
	RELEASE_OBJ( app_ );

	clear();
}

void ChordNode::clear()
{
	std::for_each( protocols_.begin(), protocols_.end(),
			ReleaseObject() );
	protocols_.clear();
}

void ChordNode::onEvent( Event *event )
{
	SearchEvent *search_event = NULL;
	switch( event->getType() ){
		case Event::SEARCH:
			search_event = static_cast<SearchEvent*>(event);
			searchKey( search_event->getKey(), search_event->getChordNetID() );
			break;
		default:
			Node::onEvent( event );
			break;
	}

}

bool ChordNode::isInNetwork( chord_net_id_t net_id )
{
	if ( !isOnline() ) return false;
	ChordProtocol *p = getProtocolByID( net_id );
	if ( NULL==p ) return false;
	if ( !(p->isActive()) ) return false;
	return true;
}

void ChordNode::searchKey( chord_key_t key, chord_net_id_t net_id )
{
	if ( !isOnline() ) {
		PeerLoger::instance().logResult(0, "query %u hit:3 hops:-1", key);
		return;
	}
	if ( !ChordGod::instance().isKeyOnline( net_id, key ) ){
		PeerLoger::instance().logResult(0, "query %u hit:2 hops:-1", key);
		return;
	}
	ChordProtocol *p = getProtocolByID(net_id);
	if ( NULL==p ){
		assert( config.scheduler_.merge_.start_time>0 ); //must be merging
		PeerLoger::instance().logResult(0, "query %u hit:0 hops:-1", key);
		return;
	}
	p->generateQuery(key);
}

void ChordNode::onReceived( Packet *packet )
{
	assert_or_ret( packet!=NULL );
	if ( !isOnline() ) return;

	PeerLoger::instance().logTrans( 0, "%d recv packet from %d", 
			getNodeID(), packet->getSrcID() );
	switch( packet->getPackType() ){
		case Packet::DATA_PACKET:
			assert(false);
			break;
		case Packet::CHORD_PACKET:
			processMessage( static_cast<ChordMsg*>(packet) );
			break;
	}

}

void ChordNode::onJoining()
{
	assert_or_ret( invalid_chord_net_id!=startup_net_id_ );
	assert_or_ret( protocols_.empty() );
	PeerLoger::instance().logDebug( 0, " node:%u join the network %u", getNodeID(), startup_net_id_ );

	Node::onJoining();

	ChordGod::instance().nodeOnline(this);

	NEW_VAR_OBJECT( ChordProtocol, p, (this, startup_net_id_) );
	protocols_.push_back( p );

	p->join();

}

void ChordNode::onDeparture()
{
	PeerLoger::instance().logDebug( 0, " node:%d leave the network", getNodeID() );
	ChordGod::instance().nodeOffline(this);
	TChordProtocolList::iterator itr = protocols_.begin(),
		itre = protocols_.end();
	for ( ; itr!=itre; itr++ ){
		ChordProtocol *p = *itr;
		p->leave();
	}
	//clear();
	Node::onDeparture();
}

bool ChordNode::merge( chord_net_id_t from, chord_net_id_t to )
{
	ChordProtocol *p_from = getProtocolByID( from );
	ChordProtocol *p_to   = getProtocolByID( to );

	assert_or_retv( p_from!=NULL, false );
	assert_or_retv( p_to==NULL, false );

	//ChordNode *s_n = p_from->getSuccessorNode();
	//if ( NULL==s_n ){
		//fprintf( stderr, "node:%u:%u moving out failed: no successor.", getChordNodeID(), getNodeID() );
		//return false;
	//}

	NEW_VAR_OBJECT( ChordProtocol, p, (this, to) );
	protocols_.push_back( p );

	p->setMerging();
	p->setOldNetID( p_from->getChordNetID() );

	p->join();
}

ChordProtocol* ChordNode::getProtocolByID( chord_net_id_t id )
{
	TChordProtocolList::iterator itr = protocols_.begin(),
		itre = protocols_.end();
	while(itr!=itre){
		ChordProtocol *p = *itr;
		if ( p->getChordNetID()==id )
			return p;
		itr++;
	}
	
	return NULL;
}

void ChordNode::processMessage( ChordMsg *msg )
{
	assert_or_ret( NULL!=msg );
	ChordProtocol *p = getProtocolByID( msg->getChordNetID() );
	assert_or_ret( NULL!=p );
	p->processMessage( msg );
}

